package daos;

import java.io.IOException;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

import model.Cliente;
import model.Persona;
import model.Vendedor;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

public class DAOPersona {
	
	private static DateFormat format = new SimpleDateFormat("dd-MM-yyyy");

	public HTable getTable() throws IOException{
		Configuration conf = HBaseConfiguration.create();
		HTable hTable = new HTable(conf, "persona");
		return hTable;
	}
	
	public void cerrarTabla(HTable tabla){
		try {
			tabla.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}
	
	public void poner(Put putRow){
		HTable tabla=null;
	try {
			tabla = this.getTable();
			tabla.put(putRow);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}finally{
			this.cerrarTabla(tabla);
		}
		
	}
	
	
	public void agregarPersona(Persona p, Put putRow){
		
		 putRow.add(Bytes.toBytes("datos"), Bytes.toBytes("fnacimiento"), Bytes.toBytes(format.format(p.getFechaNacimiento())));
		 putRow.add(Bytes.toBytes("datos"), Bytes.toBytes("nombre"), Bytes.toBytes(p.getNombre()));
		 putRow.add(Bytes.toBytes("datos"), Bytes.toBytes("apellido"), Bytes.toBytes(p.getApellido()));
		 putRow.add(Bytes.toBytes("datos"), Bytes.toBytes("mail"), Bytes.toBytes(p.getMail()));		 
		 
	}
		
	public void agregarVendedor(Vendedor v) {
		Put putRow = new Put(Bytes.toBytes(v.getLegajo()));
		this.agregarPersona(v, putRow);
		putRow.add(Bytes.toBytes("datos"), Bytes.toBytes("cargo"), Bytes.toBytes(v.getCargo()));		 
		putRow.add(Bytes.toBytes("datos"), Bytes.toBytes("sueldo"), Bytes.toBytes(v.getSueldo()));		 
		
		this.poner(putRow);
	}
		
		public void agregarCliente(Cliente c){
			Put putRow = new Put(Bytes.toBytes(c.getLegajo()));
			this.agregarPersona(c, putRow);
			putRow.add(Bytes.toBytes("datos"), Bytes.toBytes("deuda"), Bytes.toBytes(c.getDeuda()));		 
			putRow.add(Bytes.toBytes("datos"), Bytes.toBytes("categoria"), Bytes.toBytes(c.getCategoria()));		 
			
		this.poner(putRow);
		
	}
	
	public Persona obtenerIndividuo(Integer legajo){
		Result res = null ;
		try {
			res = this.getTable().get(new Get(Bytes.toBytes(legajo)));
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		Persona ind = null;
				
		if(res.containsColumn(Bytes.toBytes("datos"), Bytes.toBytes("deuda"))){
			ind = new Cliente();
			System.out.println((res.getValue(Bytes.toBytes("datos"), Bytes.toBytes("deuda"))));
			((Cliente)ind).setDeuda(Bytes.toDouble((res.getValue(Bytes.toBytes("datos"), Bytes.toBytes("deuda")))));
			((Cliente)ind).setCategoria(Bytes.toString(res.getValue(Bytes.toBytes("datos"), Bytes.toBytes("categoria"))));
		}else{
			ind = new Vendedor();
			((Vendedor)ind).setSueldo(Bytes.toDouble(res.getValue(Bytes.toBytes("datos"), Bytes.toBytes("sueldo"))));
			((Vendedor)ind).setCargo(Bytes.toString(res.getValue(Bytes.toBytes("datos"), Bytes.toBytes("cargo"))));
		}
		ind.setApellido(Bytes.toString(res.getValue(Bytes.toBytes("datos"), Bytes.toBytes("apellido"))));
		ind.setNombre(Bytes.toString(res.getValue(Bytes.toBytes("datos"), Bytes.toBytes("nombre"))));
		ind.setMail(Bytes.toString(res.getValue(Bytes.toBytes("datos"), Bytes.toBytes("mail"))));
		try {
			ind.setFechaNacimiento(format.parse(Bytes.toString(res.getValue(Bytes.toBytes("datos"), Bytes.toBytes("fnacimiento")))));
		} catch (ParseException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		return ind;		
	}

}
